package com.ipay.wechat.service.impl;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import com.ipay.utils.StringUtil;
import com.ipay.wechat.service.MessagePushService;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;

import org.json.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.ipay.kafka.KafkaProducer;
import com.ipay.redis.JedisExecutor;
import com.ipay.redis.JedisKeyPrefix;
import com.ipay.utils.Constant;
import com.ipay.utils.RandomStringUtils;
import com.ipay.wechat.component.Wechat;
import com.ipay.wechat.dao.MerchantInfoDao;
import com.ipay.wechat.dao.MessagePushDao;
import com.ipay.wechat.pojo.AccessToken;
import com.ipay.wechat.pojo.MerchantInfoEntity;
import com.ipay.wechat.pojo.MessagePush;
import com.ipay.wechat.service.MerchantInfoService;
import com.ipay.wechat.service.TradeRecordService;
import com.ipay.wechat.util.AccessTokenUtil;
import com.ipay.wechat.util.HttpUtils;
import com.ipay.wechat.util.PushMessage;
import org.springframework.web.context.WebApplicationContext;
import org.springframework.web.context.support.WebApplicationContextUtils;

@Service("merchantInfoService")
public class MerchantInfoServiceImpl implements MerchantInfoService {
    @Autowired
    private MerchantInfoDao merchantInfoDao;
    private ConsumerConnector consumer;
    @Autowired
    private Wechat wechat;
    @Autowired
    private MessagePushService messagePushService;
    @Autowired
    private JedisExecutor jedies;
    @Autowired
    private TradeRecordService tradeRecordService;

    @Override
    public MerchantInfoEntity queryObject(String id) {
        return merchantInfoDao.queryObject(id);
    }

    @Override
    public MerchantInfoEntity queryObjectStatus(String id) {
        return merchantInfoDao.queryObjectStatus(id);
    }
    @Override
    public MerchantInfoEntity querryInfoByMInfoOpenid(String id) {
        return merchantInfoDao.querryInfoByMInfoOpenid(id);
    }

    @Override
    @Deprecated
    public MerchantInfoEntity queryMInfoByMerchantNo(String id) {
        return merchantInfoDao.queryMInfoByMerchantNo(id);
    }

    @Override
    public List<MerchantInfoEntity> queryList(Map<String, Object> map) {
        return merchantInfoDao.queryList(map);
    }

    @Override
    public List<MerchantInfoEntity> queryTwoLevelList(Map<String, Object> map) {
        return merchantInfoDao.queryTwoLevelList(map);
    }

    @Override
    public int queryTotal(Map<String, Object> map) {
        return merchantInfoDao.queryTotal(map);
    }

    @Override
    public void save(MerchantInfoEntity merchantInfo) {
        merchantInfoDao.save(merchantInfo);
    }

    @Override
    public void update(MerchantInfoEntity merchantInfo) {
        merchantInfoDao.update(merchantInfo);
    }

    @Override
    public void delete(Long id) {
        merchantInfoDao.delete(id);
    }

    @Override
    public void deleteBatch(Long[] ids) {
        merchantInfoDao.deleteBatch(ids);
    }

    @Override
    public void kafkaConsumer() {
        Properties props = new Properties();
        // zookeeper 配置
        // props.put("zookeeper.connect", "127.0.0.1:2181");
        props.put("zookeeper.connect", Constant.ZOOKEEPER_CONTENT);
        // group 代表一个消费组
        props.put("group.id", Constant.GROUP_ID);
        // zk连接超时
        props.put("zookeeper.session.timeout.ms", "4000");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("rebalance.max.retries", "5");
        props.put("rebalance.backoff.ms", "1200");
        props.put("auto.offset.reset", "smallest");
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "smallest");
        // 序列化类
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        ConsumerConfig config = new ConsumerConfig(props);
        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        //每个topic使用多少个kafkastream读取, 多个consumer
        topicCountMap.put(Constant.KAFKA_TOPIC, 1);
        Map<String, List<KafkaStream<byte[], byte[]>>> msgStreams = consumer.createMessageStreams(topicCountMap);
        List<KafkaStream<byte[], byte[]>> msgStreamList = msgStreams.get(Constant.KAFKA_TOPIC);
        //使用ExecutorService来调度线程
        ExecutorService executor = Executors.newFixedThreadPool(1);
        for (int i = 0; i < msgStreamList.size(); i++) {
            KafkaStream<byte[], byte[]> kafkaStream = msgStreamList.get(i);
            executor.submit(new HanldMessageThread(kafkaStream, i));
        }
        //关闭consumer
        try {
            Thread.sleep(20000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        /*if (consumer != null) {
            consumer.shutdown();
        }
        if (executor != null) {
            executor.shutdown();
        }*/
        try {
            if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS)) {
                System.out.println("Timed out waiting for consumer threads to shut down, exiting uncleanly");
            }
        } catch (InterruptedException e) {
            System.out.println("Interrupted during shutdown, exiting uncleanly");
        }
        //consume();
    }

    /**
     * 具体处理message的线程
     *
     * @author Administrator
     */
    class HanldMessageThread implements Runnable {
        private KafkaStream<byte[], byte[]> kafkaStream = null;
        private int num;

        public HanldMessageThread(KafkaStream<byte[], byte[]> kafkaStream, int num) {
            super();
            this.kafkaStream = kafkaStream;
            this.num = num;
        }

        @Override
        public void run() {
            String access_token = tradeRecordService.getAccessToken();
            ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
            while (iterator.hasNext()) {
                byte[] mesage = iterator.next().message();
                System.out.println("consume:----mesage--" +mesage+  "------");
                String id = new String(mesage);
                System.out.println("consume:----access_token--" + access_token + "----------id------" + id + "------");
                try {
                    if (StringUtil.isNotEmpty(id)) {
                        MessagePush msg = messagePushService.queryObject(id.trim());
                        if (msg != null) {
                            System.out.println("msg:----------------mobile----------------------" + msg.getMobilePhone());
                            HashMap<String, String> map = new HashMap<String, String>();
                            map.put("index", "1");
                            map.put("openid", msg.getOpenid());

                            SimpleDateFormat format1 = new SimpleDateFormat("yyyy年MM月dd日 ");
                            String nowDate = format1.format(new Date());

							/*map.put("openid", "oejR50lPUQq5eRVL31sSIyEo0Ybg");
							map.put("first", "入驻审核结果通知");*/
                            //map.put("keyword2", "您好，您已成功通过了商户申请您的商户邀请码为："+RandomStringUtils.randomNumeric(6));

                            map.put("tempCode", wechat.getTEMPLATE1());
                            map.put("first", msg.getTitle());
                            map.put("keyword1", nowDate);
                            map.put("keyword2", msg.getContent());
                            map.put("remark", "");
                            map.put("sendType", "0");

                            String message = PushMessage.createMsgJson(map);
                            String result = HttpUtils.requestHttp("https://api.weixin.qq.com/cgi-bin/message/template/send?access_token=" + access_token, "post", message);
                            System.out.println("result:--------------------------------------" + result);
                            JSONObject jsonObj = new JSONObject(result);
                            int errCode = (int) jsonObj.get("errcode");
							/*if(errCode == 0){
								msg.setStatus(1);
								messagePushDao.update(msg);
							}*/
                            System.out.println("consume:----------------end----------------------");
                        } else {
                            System.out.println("信息不能为空");
                        }
                    } else {
                        System.out.println("信息不能为空");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

    }


    /**
     * 消费者
     */
    @Override
    public void consume() {
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(Constant.KAFKA_TOPIC, 1);

        StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
        StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());

        Map<String, List<KafkaStream<String, String>>> consumerMap = consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
        KafkaStream<String, String> stream = consumerMap.get(Constant.KAFKA_TOPIC).get(0);
        ConsumerIterator<String, String> it = stream.iterator();

        AccessToken accessToken = AccessTokenUtil.freshAccessToken(wechat);

        //String access_token = tradeRecordService.getAccessToken();
        String access_token = accessToken.getAccess_token();

        while (it.hasNext()) {
            String id = it.next().message();
            System.out.println("consume:----------------start----------------------" + id);

            try {
                MessagePush msg = messagePushService.queryObject(id);

                if (msg != null) {
                    HashMap<String, String> map = new HashMap<String, String>();
                    map.put("index", "1");
                    map.put("openid", msg.getOpenid());

                    SimpleDateFormat format1 = new SimpleDateFormat("yyyy年MM月dd日 ");
                    String nowDate = format1.format(new Date());
					
					/*map.put("openid", "oejR50lPUQq5eRVL31sSIyEo0Ybg");
					map.put("first", "入驻审核结果通知");*/
                    map.put("keyword2", "您好，您已成功通过了商户申请您的商户邀请码为：" + RandomStringUtils.randomNumeric(6));

                    map.put("tempCode", wechat.getTEMPLATE1());
                    map.put("first", msg.getTitle());
                    map.put("keyword1", nowDate);
                    //map.put("keyword2", msg.getContent());
                    map.put("remark", "");
                    map.put("sendType", "0");

                    String message = PushMessage.createMsgJson(map);
                    String result = HttpUtils.requestHttp("https://api.weixin.qq.com/cgi-bin/message/template/send?access_token=" + access_token, "post", message);
                    JSONObject jsonObj = new JSONObject(result);
                    int errCode = (int) jsonObj.get("errcode");
					/*if(errCode == 0){
						msg.setStatus(1);
						messagePushDao.update(msg);
					}*/
                    System.out.println("consume:----------------end----------------------");
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

}
